package com.zhang.hadoop.flink.test4;

import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author: zhang yufei
 * @createTime:2022/5/22 18:35
 * @description:
 */
public class SinkToFileTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        List<Event> events = new ArrayList<>();
        events.add(new Event("huichao", "./yindao", 1000L));
        events.add(new Event("yanghui", "./yindao", 1000L));
        events.add(new Event("yanghui", "./gangmen", 2000L));
        events.add(new Event("yanghui", "./siwajiao", 3000L));
        events.add(new Event("yuping", "./siwajiao", 2000L));
        events.add(new Event("yuping", "./yindao", 4000L));
        events.add(new Event("yangdan", "./gangmen", 2000L));
        events.add(new Event("yangdan", "./yindao", 6000L));
        events.add(new Event("jingru", "./niaodao", 2000L));
        DataStreamSource<Event> stream = env.fromCollection(events);

        StreamingFileSink<String> streamingFileSink = StreamingFileSink.<String>forRowFormat(new Path("./flink/output"),
                new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy(DefaultRollingPolicy.builder()
                        .withMaxPartSize(1024*1024*1024)
                        .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                        .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                        .build())
                .build();

        stream.map(data->data.toString())
                .addSink(streamingFileSink);

        env.execute();
    }
}
